/*
 
    Copyright (c) 2010 Ralf Westphal <info@ralfw.de>
     
    This file is part of NZMQ, nzmq.codeplex.com
     
    NZMQ is free software; you can redistribute it and/or modify it under
    the terms of the Lesser GNU General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.
     
    NZMQ is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    Lesser GNU General Public License for more details.
     
    You should have received a copy of the Lesser GNU General Public License
    along with this program. If not, see <http://www.gnu.org/licenses/>.

*/
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace nzmq
{
    public class ObservableSocket<T> : IObservable<T>, IDisposable
    {
        private readonly Subject<T> _observable;

        private readonly Task _polling;
        private readonly CancellationTokenSource _cancelPolling;


        internal ObservableSocket(NzmqContext context, string addressOfPublisher)
        {
            _observable = new Subject<T>();

            _cancelPolling = new CancellationTokenSource();
            _polling = Task.Factory.StartNew(() => {
                                                     var socket = context.CreateSubscriberSocket(addressOfPublisher);
                                                     Observe_socket(socket);
                                                   },
                                             _cancelPolling.Token);
        }


        public IDisposable Subscribe(IObserver<T> observer)
        {
            return _observable.Subscribe(observer);
        }


        public void Dispose()
        {
            _cancelPolling.Cancel();
            _polling.Dispose();
        }


        private void Observe_socket(NzmqSubscriberSocket socket)
        {
            do
            {
                if (Message_arrived(socket))
                    Pass_message_on_as_event(socket);
            } while (!_cancelPolling.Token.IsCancellationRequested);
        }

        private bool Message_arrived(NzmqBaseSocket socket)
        {
            const int MAX_WAIT_TIME_FOR_MESSAGE = 500;
            return NzmqCoordination.WaitAll(MAX_WAIT_TIME_FOR_MESSAGE, socket);
        }

        private void Pass_message_on_as_event(NzmqBaseSocket socket)
        {
            while(true)
            {
                object payload;
                if (!socket.TryReceive(out payload, false)) break;
                _observable.OnNext((T)payload);
            }
        }
    }
}